package com.mccken.demo.io.netty.support;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * @program: mccken-collect
 * @description: 用于异步的响应处理
 * @author: mccken
 * @create: 2022-04-30 20:52
 **/
public class RpcFutureResponse implements Future<RpcResponse> {

	// 存储requestId 与 RpcFutureResponse对象的映射关系
	public static final Map<String, RpcFutureResponse> MAP = new ConcurrentHashMap<>();

	// 是否处理完成，默认为false
	private boolean done = false;

	// 定义一个对象用于锁操作
	private final Object lock = new Object();

	// 响应对象
	private RpcResponse response;

	/**
	 * 在构造函数中，将当前对象存储到MAP中
	 */
	public RpcFutureResponse(RpcRequest rpcRequest) {
		MAP.put(rpcRequest.getRequestId(), this);
	}

	/**
	 * 根据请求id查询RpcFutureResponse对象
	 *
	 * @param requestId
	 * @return
	 */
	public static RpcFutureResponse getRpcFutureResponse(String requestId) {
		return MAP.get(requestId);
	}

	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		// 暂不实现
		return false;
	}

	@Override
	public boolean isCancelled() {
		// 暂不实现
		return false;
	}

	@Override
	public boolean isDone() {
		return done;
	}

	/**
	 * 获取数据，没有超时时间
	 *
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	@Override
	public RpcResponse get() throws InterruptedException, ExecutionException {
		return this.get(-1, TimeUnit.MILLISECONDS);
	}

	/**
	 * @param timeout the maximum time to wait
	 * @param unit    the time unit of the timeout argument
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	@Override
	public RpcResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException {

		if (isDone()) {
			return this.response;
		}

		synchronized (lock) {
			try {
				if (timeout < 0)  {
				    // 如果没有设置超时时间，就一直等待
					lock.wait();
				} else {
					// 如果设置超时时间，就在给定时间内进行等待
					long timeoutMillis = (TimeUnit.MILLISECONDS == unit) ? timeout : TimeUnit.MILLISECONDS.convert(timeout, unit);
					lock.wait(timeoutMillis);
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		if (isDone()) {
			return this.response;
		}
		
		throw new RuntimeException("获取数据，等待超时!");
	}
	
	public void setResponse(RpcResponse response) {
		this.response = response;
		synchronized (lock) {{
			this.done = true;
			// 唤醒等待的线程
			this.lock.notifyAll();
		}}
		
		// 删除已缓存的对象
		MAP.remove(response.getRequestId());
	}
}
